/*
 * Copyright 2012 The Netty Project
 *
 * The Netty Project licenses this file to you under the Apache License,
 * version 2.0 (the "License"); you may not use this file except in compliance
 * with the License. You may obtain a copy of the License at:
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations
 * under the License.
 */

package io.netty.buffer;

import java.nio.ByteBuffer;
import java.util.ArrayDeque;
import java.util.Deque;

/**
 * Description of algorithm for PageRun/PoolSubpage allocation from PoolChunk
 * <p>
 * Notation: The following terms are important to understand the code
 * > page  - a page is the smallest unit of memory chunk that can be allocated
 * > chunk - a chunk is a collection of pages
 * > in this code chunkSize = 2^{maxOrder} * pageSize
 * <p>
 * To begin we allocate a byte array of size = chunkSize
 * Whenever a ByteBuf of given size needs to be created we search for the first position
 * in the byte array that has enough empty space to accommodate the requested size and
 * return a (long) handle that encodes this offset information, (this memory segment is then
 * marked as reserved so it is always used by exactly one ByteBuf and no more)
 * <p>
 * For simplicity all sizes are normalized according to PoolArena#normalizeCapacity method
 * This ensures that when we request for memory segments of size >= pageSize the normalizedCapacity
 * equals the next nearest power of 2
 * <p>
 * To search for the first offset in chunk that has at least requested size available we construct a
 * complete balanced binary tree and store it in an array (just like heaps) - memoryMap
 * <p>
 * The tree looks like this (the size of each node being mentioned in the parenthesis)
 * <p>
 * depth=0        1 node (chunkSize)
 * depth=1        2 nodes (chunkSize/2)
 * ..
 * ..
 * depth=d        2^d nodes (chunkSize/2^d)
 * ..
 * depth=maxOrder 2^maxOrder nodes (chunkSize/2^{maxOrder} = pageSize)
 * <p>
 * depth=maxOrder is the last level and the leafs consist of pages
 * <p>
 * With this tree available searching in chunkArray translates like this:
 * To allocate a memory segment of size chunkSize/2^k we search for the first node (from left) at height k
 * which is unused
 * <p>
 * Algorithm:
 * ----------
 * Encode the tree in memoryMap with the notation
 * memoryMap[id] = x => in the subtree rooted at id, the first node that is free to be allocated
 * is at depth x (counted from depth=0) i.e., at depths [depth_of_id, x), there is no node that is free
 * <p>
 * As we allocate & free nodes, we update values stored in memoryMap so that the property is maintained
 * <p>
 * Initialization -
 * In the beginning we construct the memoryMap array by storing the depth of a node at each node
 * i.e., memoryMap[id] = depth_of_id
 * <p>
 * Observations:
 * -------------
 * 1) memoryMap[id] = depth_of_id  => it is free / unallocated
 * 2) memoryMap[id] > depth_of_id  => at least one of its child nodes is allocated, so we cannot allocate it, but
 * some of its children can still be allocated based on their availability
 * 3) memoryMap[id] = maxOrder + 1 => the node is fully allocated & thus none of its children can be allocated, it
 * is thus marked as unusable
 * <p>
 * Algorithm: [allocateNode(d) => we want to find the first node (from left) at height h that can be allocated]
 * ----------
 * 1) start at root (i.e., depth = 0 or id = 1)
 * 2) if memoryMap[1] > d => cannot be allocated from this chunk
 * 3) if left node value <= h; we can allocate from left subtree so move to left and repeat until found
 * 4) else try in right subtree
 * <p>
 * Algorithm: [allocateRun(size)]
 * ----------
 * 1) Compute d = log_2(chunkSize/size)
 * 2) Return allocateNode(d)
 * <p>
 * Algorithm: [allocateSubpage(size)]
 * ----------
 * 1) use allocateNode(maxOrder) to find an empty (i.e., unused) leaf (i.e., page)
 * 2) use this handle to construct the PoolSubpage object or if it already exists just call init(normCapacity)
 * note that this PoolSubpage object is added to subpagesPool in the PoolArena when we init() it
 * <p>
 * Note:
 * -----
 * In the implementation for improving cache coherence,
 * we store 2 pieces of information depth_of_id and x as two byte values in memoryMap and depthMap respectively
 * <p>
 * memoryMap[id]= depth_of_id  is defined above
 * depthMap[id]= x  indicates that the first node which is free to be allocated is at depth x (from root)
 */
final class PoolChunk<T> implements PoolChunkMetric {

    private static final int INTEGER_SIZE_MINUS_ONE = Integer.SIZE - 1;
    /**
     * 当前PoolChunk属于那个Arena
     */
    final PoolArena<T> arena;
    /**
     * 管理的内存，一般是16mb大小
     * 如果我们使用的是直接内存：那这块内存就是ByteBuffer而ByteBufer又是Unsafe类为我们直接分配的内存
     * 如果我们使用的是堆内存：这个属性就是一个byte数组
     */
    final T memory;
    /**
     * 这个也是个内存偏移量，默认值为0
     */
    final int offset;
    /**
     * 判断当前的内存块是否可以被池化管理
     * true：是非池化的
     * false：是池化的
     */
    final boolean unpooled;
    /**
     * 一个page的大小默认为8192
     */
    private final int pageSize;
    /**
     * 13，1左移13位后可以得到一个page的大小
     */
    private final int pageShifts;
    /**
     * 一个Chunk树的最大深度位11
     */
    private final int maxOrder;
    /**
     * 一个Chunk的大小默认为16mb
     */
    private final int chunkSize;
    /**
     * Chunk的满二叉树的：可分配深度分配能力值数组
     * 对应的索引的树节点，如果管理的内存被划分出去后该值会改变
     */
    private final byte[] memoryMap;
    /**
     * 表示当前索引的树节点所在树的深度是多少，注意初始化后不会改变
     */
    private final byte[] depthMap;
    /**
     * Chunk内存块的每一个叶子节点的大小都是8KB，这8KB就可以被一个PoolSubpage包装
     * 而这个PoolSubpage数组的容量就是2048。其实就是Chunk每分配出一个叶子结点大小的内存，就会使用一个PoolSubpage来包装一下
     */
    private final PoolSubpage<T>[] subpages;
    /**
     * Used to determine if the requested capacity is equal to or greater than pageSize.
     * 1111 1111 1111 1111 1110 0000 0000 0000
     * 当从Chunk申请大于等于一个page的时候，申请的容量与mask按位与会得到一个非0值
     */
    private final int subpageOverflowMask;
    /**
     * 24
     */
    private final int log2ChunkSize;
    /**
     * 最多能申请多少个subPage，2048
     */
    private final int maxSubpageAllocs;
    /**
     * 用于将内存标记为不可用，默认值为 maxOrder+1 = 12
     * Chunk是使用一颗满二叉树来表示ByteBuffer的占用情况的，当二叉树的叶子节点被分配出去了那么可分配的能力值赋值为12也就是已经分配出去了
     */
    private final byte unusable;
    /**
     * 我们上面说了，如果使用的是直接内存，会创建ByteBuffer，而ByteBuffer底层也是用Unsafe来分配的内存
     * 所以这里会对创建的ByteBuffer用队列做一个缓存
     * 这里缓存的ByteBuffer是被ByteBuf中用到的
     * 类型是ArrayDeque
     */
    private final Deque<ByteBuffer> cachedNioBuffers;
    /**
     * 当前Chunk还可以分配的字节数量：16mb
     * 会随着分配出去之后逐渐减少
     */
    private int freeBytes;
    /**
     * 该Chunk属于哪个PoolChunkList
     */
    PoolChunkList<T> parent;
    /**
     * 该Chunk内存块在PoolChunkList中的前驱节点
     */
    PoolChunk<T> prev;
    /**
     * 该Chunk内存块在PoolChunkList中的下一个节点
     */
    PoolChunk<T> next;


    /**
     * @param arena      Arena对象
     * @param memory     ByteBuffer
     * @param pageSize   8192
     * @param maxOrder   11
     * @param pageShifts 13
     * @param chunkSize  16mb
     * @param offset     0
     */
    PoolChunk(PoolArena<T> arena, T memory, int pageSize, int maxOrder, int pageShifts, int chunkSize, int offset) {
        unpooled = false;
        this.arena = arena;
        this.memory = memory;
        this.pageSize = pageSize;
        this.pageShifts = pageShifts;
        this.maxOrder = maxOrder;
        this.chunkSize = chunkSize;
        this.offset = offset;
        unusable = (byte) (maxOrder + 1);
        log2ChunkSize = log2(chunkSize);
        subpageOverflowMask = ~(pageSize - 1);
        freeBytes = chunkSize;

        assert maxOrder < 30 : "maxOrder should be < 30, but is: " + maxOrder;
        maxSubpageAllocs = 1 << maxOrder;
        // 创建一个长度为4096的数组
        memoryMap = new byte[maxSubpageAllocs << 1];
        depthMap = new byte[memoryMap.length];
        int memoryMapIndex = 1;
        for (int d = 0; d <= maxOrder; ++d) {
            int depth = 1 << d;
            for (int p = 0; p < depth; ++p) {
                memoryMap[memoryMapIndex] = (byte) d;
                depthMap[memoryMapIndex] = (byte) d;
                memoryMapIndex++;
            }
        }
        // 创建PoolSubpage数组长度为2048
        subpages = newSubpageArray(maxSubpageAllocs);
        cachedNioBuffers = new ArrayDeque<>(8);
    }

    /**
     * Creates a special chunk that is not pooled.
     */
    PoolChunk(PoolArena<T> arena, T memory, int size, int offset) {
        unpooled = true;
        this.arena = arena;
        this.memory = memory;
        this.offset = offset;
        memoryMap = null;
        depthMap = null;
        subpages = null;
        subpageOverflowMask = 0;
        pageSize = 0;
        pageShifts = 0;
        maxOrder = 0;
        unusable = (byte) (maxOrder + 1);
        chunkSize = size;
        log2ChunkSize = log2(chunkSize);
        maxSubpageAllocs = 0;
        cachedNioBuffers = null;
    }

    @SuppressWarnings("unchecked")
    private PoolSubpage<T>[] newSubpageArray(int size) {
        return new PoolSubpage[size];
    }

    @Override
    public int usage() {
        final int freeBytes;
        synchronized (arena) {
            freeBytes = this.freeBytes;
        }
        return usage(freeBytes);
    }

    private int usage(int freeBytes) {
        if (freeBytes == 0) {
            return 100;
        }

        int freePercentage = (int) (freeBytes * 100L / chunkSize);
        if (freePercentage == 0) {
            return 99;
        }
        return 100 - freePercentage;
    }

    /**
     *
     * @param buf ByteBuf
     * @param reqCapacity 请求的内存大小
     * @param normCapacity 规格后的内存大小
     * @return
     */
    boolean allocate(PooledByteBuf<T> buf, int reqCapacity, int normCapacity) {
        // 分配内存占用的树节点的id，可能是-1，-1表示申请失败
        final long handle;
        // 条件成立：normCapacity是一个大于等于8192的内存
        if ((normCapacity & subpageOverflowMask) != 0) {
            // 分配一个page以上，返回的是PoolChunk中可以分配的节点id
            handle = allocateRun(normCapacity);
        } else {
            // 说明需要的是小规格的内存也就是subPage
            handle = allocateSubpage(normCapacity);
        }
        // 走到这里说明从该Chunk中分配内存失败了，所以就会寻找其他的Chunk内存块申请内存
        // 所谓内存分配失败，就意味着该Chunk内存块中没有足够的内存可分配出去了。
        if (handle < 0) {
            return false;
        }
        // 还记得之前在讲解该类的成员变量时有一个Deque<ByteBuffer> cachedNioBuffers队列吗？
        // 这个队列缓存了一些ByteBuffer，这些ByteBuffer是要给PooledDirectByteBuf内部使用的。这里缓存一些其实就是避免频繁创建
        // 然后频繁销毁，减少垃圾回收而已
        // 如果队列中没有则直接返回null
        ByteBuffer nioBuffer = cachedNioBuffers != null ? cachedNioBuffers.pollLast() : null;
        // 初始化Bytebuf，这个Bytebuf就是从方法一开始就传进来的PooledByteBuf
        // 这里也在一次印证了分配的内存其实是交给了ByteBuf，或者说是让ByteBuf做了一层包装才能被使用的
        initBuf(buf, nioBuffer, handle, reqCapacity);
        return true;
    }

    /**
     * Update method used by allocate
     * This is triggered only when a successor is allocated and all its predecessors
     * need to update their state
     * The minimal depth at which subtree rooted at id has some free space
     *
     * @param id id
     */
    private void updateParentsAlloc(int id) {
        while (id > 1) {
            int parentId = id >>> 1;
            byte val1 = value(id);
            byte val2 = value(id ^ 1);
            byte val = val1 < val2 ? val1 : val2;
            setValue(parentId, val);
            id = parentId;
        }
    }

    /**
     * Update method used by free
     * This needs to handle the special case when both children are completely free
     * in which case parent be directly allocated on request of size = child-size * 2
     *
     * @param id id
     */
    private void updateParentsFree(int id) {
        int logChild = depth(id) + 1;
        while (id > 1) {
            int parentId = id >>> 1;
            byte val1 = value(id);
            byte val2 = value(id ^ 1);
            logChild -= 1; // in first iteration equals log, subsequently reduce 1 from logChild as we traverse up

            if (val1 == logChild && val2 == logChild) {
                setValue(parentId, (byte) (logChild - 1));
            } else {
                byte val = val1 < val2 ? val1 : val2;
                setValue(parentId, val);
            }

            id = parentId;
        }
    }

    /**
     * Algorithm to allocate an index in memoryMap when we query for a free node
     * at depth d
     *
     * @param d depth
     * @return index in memoryMap
     */
    private int allocateNode(int d) {
        // 从根节点开始分配内存
        int id = 1;
        // -2048
        int initial = -(1 << d);
        // 获取Chunk二叉树根节点的可分配深度能力值
        byte val = value(id);
        // 如果根节点的可分配深度能力值 > 需要的深度，说明Chunk剩余的内存不足以支撑本次内容的申请，返回-1申请失败
        if (val > d) {
            return -1;
        }
        /*
         * 深度优先的搜索算法
         * 条件1：可分配深度能力值 < 需要的深度 说明当前节点管理的内存支持这次申请
         * 条件2：如果等于0说明现在还没有查找到第11层
         */
        while (val < d || (id & initial) == 0) {
            // 继续向左下搜索
            id <<= 1;
            // 获取当前节点的可分配深度能力值
            val = value(id);
            // 如果不足以就看看这个节点旁边的伙伴支不支持
            if (val > d) {
                // 伙伴算法获取兄弟节点
                id ^= 1;
                val = value(id);
            }
        }
        // 获取查询出来的合适分配节点的深度能力值
        byte value = value(id);
        assert value == d && (id & initial) == 1 << d : String.format("val = %d, id & initial = %d, d = %d",
                value, id & initial, d);
        // 占用这个节点
        setValue(id, unusable);
        // 更新父辈的分配深度能力值
        updateParentsAlloc(id);
        return id;
    }

    /**
     * Allocate a run of pages (>=1)
     *
     * @param normCapacity normalized capacity
     * @return index in memoryMap
     */
    private long allocateRun(int normCapacity) {
        // 申请内存对应的满二叉树对应的深度值，比如8kb就是11
        int d = maxOrder - (log2(normCapacity) - pageShifts);
        // 到满二叉树上查找到对应的可以分配的节点id
        int id = allocateNode(d);
        // 说明当前Chunk不够分配的
        if (id < 0) {
            return id;
        }
        // 更新freeBytes
        freeBytes -= runLength(id);
        return id;
    }

    /**
     * Create / initialize a new PoolSubpage of normCapacity
     * Any PoolSubpage created / initialized here is added to subpage pool in the PoolArena that owns this PoolChunk
     *
     * @param normCapacity normalized capacity
     * @return index in memoryMap
     */
    private long allocateSubpage(int normCapacity) {
        // 根据规格后大小去查找两个Subpage数组中对应的下标头节点
        PoolSubpage<T> head = arena.findSubpagePoolHead(normCapacity);
        // 获取最大深度
        int d = maxOrder;
        // head节点作为锁
        synchronized (head) {
            // 获取一个可以分配指定深度的节点的id，这里会返回最底层的page的id
            int id = allocateNode(d);
            // 如果没有可以分配的节点就返回
            if (id < 0) {
                return id;
            }
            final PoolSubpage<T>[] subpages = this.subpages;
            final int pageSize = this.pageSize;
            // 更新Chunk剩余空间
            freeBytes -= pageSize;
            // 取模算法找个下标
            int subpageIdx = subpageIdx(id);
            // 获取出对应位置的subpage一般是null
            PoolSubpage<T> subpage = subpages[subpageIdx];
            // 一般是这个
            if (subpage == null) {
                subpage = new PoolSubpage<T>(head, this, id, runOffset(id), pageSize, normCapacity);
                subpages[subpageIdx] = subpage;
            } else {
                subpage.init(head, normCapacity);
            }
            // 申请小规格内存的入口
            return subpage.allocate();
        }
    }

    /**
     * Free a subpage or a run of pages
     * When a subpage is freed from PoolSubpage, it might be added back to subpage pool of the owning PoolArena
     * If the subpage pool in PoolArena has at least one other PoolSubpage of given elemSize, we can
     * completely free the owning Page so it is available for subsequent allocations
     *
     * @param handle handle to free
     */
    void free(long handle, ByteBuffer nioBuffer) {
        // 获得两个索引
        int memoryMapIdx = memoryMapIdx(handle);
        int bitmapIdx = bitmapIdx(handle);
        // 如果bitmapIdx不是0说明要释放的是Subpage规格的
        if (bitmapIdx != 0) {
            // 根据memoryMapIdx计算下标然后拿到subpage
            PoolSubpage<T> subpage = subpages[subpageIdx(memoryMapIdx)];
            assert subpage != null && subpage.doNotDestroy;
            // 获取Arena范围该内存规格在两个PoolSubpage数组上的合适的头节点
            PoolSubpage<T> head = arena.findSubpagePoolHead(subpage.elemSize);
            // 锁主头
            synchronized (head) {
                // 调用Subpage的释放方法，参数1是head，参数2是解码后真正的bitmapIdx
                // 返回true就说明这个Subpage还能用，false就是不能用了被释放了
                if (subpage.free(head, bitmapIdx & 0x3FFFFFFF)) {
                    return;
                }
            }
        }
        // 更新freeBytes
        freeBytes += runLength(memoryMapIdx);
        // 更新节点的可分配深度能力值
        setValue(memoryMapIdx, depth(memoryMapIdx));
        // 更新所有父节点的可分配深度能力值
        updateParentsFree(memoryMapIdx);

        if (nioBuffer != null && cachedNioBuffers != null &&
                cachedNioBuffers.size() < PooledByteBufAllocator.DEFAULT_MAX_CACHED_BYTEBUFFERS_PER_CHUNK) {
            cachedNioBuffers.offer(nioBuffer);
        }
    }

    void initBuf(PooledByteBuf<T> buf, ByteBuffer nioBuffer, long handle, int reqCapacity) {
        // 通过handle的高32位和低32位分别取出memoryMapIdx和bitmapIdx
        // 我们之前说过，handle是个内存偏移量，实际上就是它的long的64为存储了它在Chunk内存块中的下标
        // 和在位图中的具体位置
        // 取出在Chunk中的数组下标
        // 获取低32位的值
        int memoryMapIdx = memoryMapIdx(handle);
        // 获取高32位也就是bitmapIdx
        int bitmapIdx = bitmapIdx(handle);
        // 这里是判断了没有使用位图，就意味着是从Chunk中分配的大于等于8KB大小的内存，所以直接初始化Buf即可
        if (bitmapIdx == 0) {
            // 获得可分配能力值
            byte val = value(memoryMapIdx);
            assert val == unusable : String.valueOf(val);
            // 初始化ByteBuf
            buf.init(this, nioBuffer, handle, runOffset(memoryMapIdx) + offset,
                    reqCapacity, runLength(memoryMapIdx), arena.parent.threadCache());
        } else {
            // 分配的是subpage才会走这里
            initBufWithSubpage(buf, nioBuffer, handle, bitmapIdx, reqCapacity);
        }
    }

    void initBufWithSubpage(PooledByteBuf<T> buf, ByteBuffer nioBuffer, long handle, int reqCapacity) {
        initBufWithSubpage(buf, nioBuffer, handle, bitmapIdx(handle), reqCapacity);
    }

    private void initBufWithSubpage(PooledByteBuf<T> buf, ByteBuffer nioBuffer,
                                    long handle, int bitmapIdx, int reqCapacity) {
        assert bitmapIdx != 0;
        // 获取低32位代表的叶子节点id
        int memoryMapIdx = memoryMapIdx(handle);
        // 获取对应的Subpage对象
        PoolSubpage<T> subpage = subpages[subpageIdx(memoryMapIdx)];
        assert subpage.doNotDestroy;
        assert reqCapacity <= subpage.elemSize;
        /*
         * 参数1：PoolChunk对象
         * 参数2：为NULL即可
         * 参数3：Handle
         * 参数4：
         *      - bitmapIdx & 0x3FFFFFFF = 解析出来真正的bitmapIdx值
         *      - bitmapIdx * subpage.elemSize = 当前bitmapIdx占用的内存在subpage上的偏移位置
         *      - offset一般是0
         */
        buf.init(
                this, nioBuffer, handle,
                runOffset(memoryMapIdx) + (bitmapIdx & 0x3FFFFFFF) * subpage.elemSize + offset,
                reqCapacity, subpage.elemSize, arena.parent.threadCache());
    }

    private byte value(int id) {
        return memoryMap[id];
    }

    private void setValue(int id, byte val) {
        memoryMap[id] = val;
    }

    private byte depth(int id) {
        return depthMap[id];
    }

    private static int log2(int val) {
        // compute the (0-based, with lsb = 0) position of highest set bit i.e, log2
        return INTEGER_SIZE_MINUS_ONE - Integer.numberOfLeadingZeros(val);
    }

    private int runLength(int id) {
        return 1 << log2ChunkSize - depth(id);
    }

    /**
     * 计算从Chunk内存块中分配出去的内存在Chunk内存块的memory中的具体偏移量
     * id是2048返回0，id是2049返回8k，id是2050返回16k，id是2051返回24k
     * @param id
     * @return
     */
    private int runOffset(int id) {
        int shift = id ^ 1 << depth(id);
        return shift * runLength(id);
    }

    private int subpageIdx(int memoryMapIdx) {
        return memoryMapIdx ^ maxSubpageAllocs; // remove highest set bit, to get offset
    }

    private static int memoryMapIdx(long handle) {
        return (int) handle;
    }

    private static int bitmapIdx(long handle) {
        return (int) (handle >>> Integer.SIZE);
    }

    @Override
    public int chunkSize() {
        return chunkSize;
    }

    @Override
    public int freeBytes() {
        synchronized (arena) {
            return freeBytes;
        }
    }

    @Override
    public String toString() {
        final int freeBytes;
        synchronized (arena) {
            freeBytes = this.freeBytes;
        }

        return new StringBuilder()
                .append("Chunk(")
                .append(Integer.toHexString(System.identityHashCode(this)))
                .append(": ")
                .append(usage(freeBytes))
                .append("%, ")
                .append(chunkSize - freeBytes)
                .append('/')
                .append(chunkSize)
                .append(')')
                .toString();
    }

    void destroy() {
        arena.destroyChunk(this);
    }
}
